package rdd

import org.apache.spark.{SparkConf, SparkContext}

object RDD_AccumulatorsExample {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("RDD_AccumulatorsExample")

    val sc = new SparkContext(conf)
    // 定义计数器
    val starCounter = sc.longAccumulator("star_counter")
    val scoreCounter = sc.longAccumulator("score_counter")
    val commCounter = sc.longAccumulator("comm_counter")

    // 加载外部数据源，构造RDD
    val rdd1 = sc.textFile("data/sample.csv")

    // 过滤掉标题行
    val rdd2 = rdd1.filter(line => !line.startsWith("SEQ"))

    // 将关键字段有缺失值的记录删除：即将字段{星级、评论数、评分}中任意字段为空的数据删除
    // 并打印输出删除条目数 - 使用计数器统计
    val rdd3 = rdd2.map(_.split(",",-1))
      .filter(arr => {
        var flag = true       // 定义标志变量
        // 如果"星级"字段为空
        if(arr(6)==null || arr(6).trim.isEmpty){
          starCounter.add(1)    // 全局计数器 + 1
          flag = false
        }

        if(arr(10)==null || arr(10).trim.isEmpty){
          scoreCounter.add(1)    // 全局计数器 + 1
          flag = false
        }

        if(arr(11)==null || arr(11).trim.isEmpty){
          commCounter.add(1)    // 全局计数器 + 1
          flag = false
        }

        flag
      })

    // 显示
    println(s"过滤前记录数${rdd2.count}，过滤后记录后${rdd3.count}")
    println(s"删除的星级字段缺失的记录数是：${starCounter.value}")
    println(s"删除的评分字段缺失的记录数是：${scoreCounter.value}")
    println(s"删除的评论数字段缺失的记录数是：${commCounter.value}")


    sc.stop()
  }
}
